package com.ls.fw.data.support.parse;

import java.util.List;

import org.apache.hadoop.hbase.client.Result;
import org.apache.solr.common.SolrInputDocument;

import com.google.common.collect.Lists;

/**
 * Extracts a {@code SolrInputDocument} from an HBase {@code Result} object.
 */
public class HBaseSolrDocumentExtractor implements SolrDocumentExtractor {

    private String fieldName;
    private ByteArrayExtractor valueExtractor;
    private ByteArrayValueMapper valueMapper;

    public HBaseSolrDocumentExtractor(String fieldName, ByteArrayExtractor valueExtractor,
            ByteArrayValueMapper valueMapper) {
        this.fieldName = fieldName;
        this.valueExtractor = valueExtractor;
        this.valueMapper = valueMapper;
    }

    /**
     * Extracts byte arrays from the given {@code Result}, and transforms them into a {@code SolrInputDocument}.
     * 
     * @param result source of byte array data
     * @param solrInputDocument document where indexable data is to be added
     */
    @Override
    public void extractDocument(Result result, SolrInputDocument solrInputDocument) {
        List<Object> values = Lists.newArrayList();
        for (byte[] bytes : valueExtractor.extract(result)) {
            values.addAll(valueMapper.map(bytes));
        }
        solrInputDocument.addField(fieldName, values);
    }

}
